/*
 * MIT License
 *
 * Copyright (c) 2023 北京凯特伟业科技有限公司
 *
 * Permission is hereby granted, free of charge, to any person obtaining a copy
 * of this software and associated documentation files (the "Software"), to deal
 * in the Software without restriction, including without limitation the rights
 * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
 * copies of the Software, and to permit persons to whom the Software is
 * furnished to do so, subject to the following conditions:
 *
 * The above copyright notice and this permission notice shall be included in all
 * copies or substantial portions of the Software.
 *
 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
 * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
 * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
 * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
 * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
 * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 * SOFTWARE.
 */
package com.je.bpm.engine.impl.asyncexecutor;

import com.je.bpm.engine.ActivitiOptimisticLockingException;
import com.je.bpm.engine.impl.persistence.entity.JobEntity;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * Runnable that checks the {@link com.je.bpm.engine.runtime.Job} entities periodically for 'expired' jobs.
 * <p>
 * When a job is executed, it is first locked (lock owner and lock time is set).
 * A job is expired when this lock time is exceeded. This can happen when an executor
 * goes down before completing a task.
 * <p>
 * This runnable will find such jobs and reset them, so they can be picked up again.
 */
public class ResetExpiredJobsRunnable implements Runnable {

    private static Logger log = LoggerFactory.getLogger(ResetExpiredJobsRunnable.class);

    protected final AsyncExecutor asyncExecutor;

    protected volatile boolean isInterrupted;
    protected final Object MONITOR = new Object();
    protected final AtomicBoolean isWaiting = new AtomicBoolean(false);

    public ResetExpiredJobsRunnable(AsyncExecutor asyncExecutor) {
        this.asyncExecutor = asyncExecutor;
    }

    @Override
    public synchronized void run() {
        log.info("{} starting to reset expired jobs");
        Thread.currentThread().setName("activiti-reset-expired-jobs");

        while (!isInterrupted) {

            try {

                List<JobEntity> expiredJobs = asyncExecutor.getProcessEngineConfiguration().getCommandExecutor()
                        .execute(new FindExpiredJobsCmd(asyncExecutor.getResetExpiredJobsPageSize()));

                List<String> expiredJobIds = new ArrayList<String>(expiredJobs.size());
                for (JobEntity expiredJob : expiredJobs) {
                    expiredJobIds.add(expiredJob.getId());
                }

                if (expiredJobIds.size() > 0) {
                    asyncExecutor.getProcessEngineConfiguration().getCommandExecutor()
                            .execute(new ResetExpiredJobsCmd(expiredJobIds));
                }
            } catch (Throwable e) {
                if (e instanceof ActivitiOptimisticLockingException) {
                    log.debug("Optmistic lock exception while resetting locked jobs", e);
                } else {
                    log.error("exception during resetting expired jobs", e.getMessage(), e);
                }
            }

            // Sleep
            try {

                synchronized (MONITOR) {
                    if (!isInterrupted) {
                        isWaiting.set(true);
                        MONITOR.wait(asyncExecutor.getResetExpiredJobsInterval());
                    }
                }

            } catch (InterruptedException e) {
                if (log.isDebugEnabled()) {
                    log.debug("async reset expired jobs wait interrupted");
                }
            } finally {
                isWaiting.set(false);
            }

        }

        log.info("{} stopped resetting expired jobs");
    }

    public void stop() {
        synchronized (MONITOR) {
            isInterrupted = true;
            if (isWaiting.compareAndSet(true, false)) {
                MONITOR.notifyAll();
            }
        }
    }


}
